package com.hkj.configcenter.core.processor;

import com.alibaba.fastjson.JSON;
import com.github.zkclient.IZkDataListener;
import com.github.zkclient.ZkClient;
import com.hkj.configcenter.core.Context;
import com.hkj.configcenter.core.model.FieldInfo;
import com.hkj.configcenter.core.utils.FieldUtils;
import com.hkj.configcenter.core.utils.ZkUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
public class ZkProcessor implements Processor {

    private ZkClient zkClient;

    private String versionPath;

    private String versionValue;

    ZkProcessor() {
        initConnection();
        initListener();
    }

    private void initConnection() {
        zkClient = ZkUtils.getConnection(Context.getZkUrl());
        Context.zkClient = zkClient;

        if (!zkClient.exists(Context.getResourceConfigPath())) {
            zkClient.createPersistent(Context.getResourceConfigPath(), true);
        }

        if (!zkClient.exists(Context.getFieldConfigPath())) {
            zkClient.createPersistent(Context.getFieldConfigPath(), true);
        }

        if (!zkClient.exists(Context.getOnlineMachinesPath())) {
            zkClient.createPersistent(Context.getOnlineMachinesPath(), "1".getBytes());
        }

        if (!zkClient.exists(Context.getCommonConfigPath())) {
            zkClient.createPersistent(Context.getCommonConfigPath(), true);
        }

        versionValue = ZkUtils.readData(zkClient, Context.getOnlineMachinesPath());
        versionPath = zkClient.createEphemeralSequential(Context.getOnlineMachinesPath() + "/", ZkUtils.getVersionValue(versionValue).getBytes());
        log.info("【配置中心】 VersionPath ==> {}", versionPath);
    }

    private void initListener() {

        zkClient.subscribeChildChanges(Context.getOnlineMachinesPath(), (parentPath, currentChildren) -> {
            log.info(versionPath + " ==> " + currentChildren);
            List<String> children = zkClient.getChildren(parentPath);

            if (children == null) {
                log.info("【配置中心】 - ParentVersionPath 被删除 ==> {}", parentPath);
                return;
            }

            boolean rebuild = true;
            for (String child : children) {
                if (child.equals(getKey(versionPath))) {
                    rebuild = false;
                }
            }

            if (rebuild) {
                versionPath = zkClient.createEphemeralSequential(Context.getOnlineMachinesPath() + "/", ZkUtils.getVersionValue(versionValue).getBytes());
                log.info("【配置中心】 - VersionPath 重新连接 ==> {}", versionPath);
            }
        });

        zkClient.subscribeDataChanges(Context.getOnlineMachinesPath(), new IZkDataListener() {

            @Override
            public void handleDataChange(String parentVersionPath, byte[] versionData) {
                versionValue = new String(versionData);
                log.info("【配置中心】 ParentVersionPath 数据修改 data = {}, path ==> {}", versionValue, parentVersionPath);

                Map<String, String> map = new HashMap<>();
                map.put("date", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                for (Map.Entry<String, List<FieldInfo>> entry : Context.getFieldInfoCache().entrySet()) {

                    String path = entry.getKey();
                    List<FieldInfo> fieldInfos = entry.getValue();
                    String newValue = FieldUtils.getUnremarkValue(ZkUtils.readData(zkClient, path));

                    fieldInfos.forEach(fieldInfo -> {
                        if (StringUtils.isEmpty(newValue) || !newValue.equals(fieldInfo.getValue())) {
                            String oldValue = fieldInfo.getValue();
                            fieldInfo.setValue(newValue);
                            FieldUtils.set(fieldInfo.getInstance(), fieldInfo.getField(), newValue);
                            map.put(fieldInfo.getInstance().getClass().getSimpleName() + "." + fieldInfo.getField().getName(), newValue);
                            log.info("【配置中心】 - 参数修改 {} = {} , 旧 = {}", path, newValue, oldValue);
                        }
                    });
                }

                zkClient.writeData(versionPath, (ZkUtils.getVersionValue(versionValue) + Context.SPLIT_FLAG + JSON.toJSONString(map)).getBytes());
            }

            public void handleDataDeleted(String dataPath) {
                log.info("【配置中心】 - ParentVersionPath 节点被异常删除，重新恢复");
                if (!zkClient.exists(Context.getOnlineMachinesPath())) {
                    zkClient.createPersistent(Context.getOnlineMachinesPath(), "1".getBytes());
                }
            }

        });

        log.info("【配置中心】 - 开始监听ParentVersionPath and VersionPath 节点变化 ==> {}", Context.getOnlineMachinesPath());
    }

    public void processorResourceConfig() {

        Map<String, String> commonMap = ZkUtils.getChildrenValue(zkClient, Context.getCommonConfigPath());

        for (Map.Entry<String, String> entry : commonMap.entrySet()) {
            String value = FieldUtils.getUnremarkValue(entry.getValue());
            String pathName = getKey(entry.getKey());
            String path_value = entry.getKey() + Context.SPLIT_FLAG + value;

            Context.getField_PathValue().put(pathName, path_value);
            System.setProperty(pathName, value);
            log.info("【配置中心】【Common参数】 - {}={}", entry.getKey(), value);
        }

        Map<String, String> resourceMap = ZkUtils.getChildrenValue(zkClient, Context.getResourceConfigPath());
        for (Map.Entry<String, String> entry : resourceMap.entrySet()) {
            String key = getKey(entry.getKey());
            String value = FieldUtils.getUnremarkValue(entry.getValue().trim());
            if (Context.getField_PathValue().containsKey(key)) {
                log.info("【配置中心】【Resource参数】 - 覆盖Common参数【{}】{}==>{}", key, Context.getField_PathValue().get(key).split(Context.SPLIT_FLAG)[0], value);
            }

            System.setProperty(key, value);
            log.info("【配置中心】【Resource参数】 - {}={}", key, value);
        }
    }

    public void processorFieldConfig() {

        Map<String, String> fieldMap = ZkUtils.getChildrenValue(zkClient, Context.getFieldConfigPath());

        for (Map.Entry<String, String> entry : fieldMap.entrySet()) {
            String value = FieldUtils.getUnremarkValue(entry.getValue());
            String pathName = getKey(entry.getKey());
            String path_value = entry.getKey() + Context.SPLIT_FLAG + value;
            if (Context.getField_PathValue().containsKey(pathName)) {
                log.info("【配置中心】【Field参数】 - 覆盖Common参数【{}】{}==>{}", pathName, Context.getField_PathValue().get(pathName).split(Context.SPLIT_FLAG)[1], value);
            }

            Context.getField_PathValue().put(pathName, path_value);
            System.setProperty(pathName, value);
            log.info("【配置中心】【Field参数】 - {}={}", entry.getKey(), value);
        }

    }

    public static String getKey(String path) {
        return path.substring(path.lastIndexOf("/") + 1, path.length()).trim();
    }

    @Override
    public void destroy() {
        if (zkClient != null) {
            zkClient.close();
            log.info("[ZK资源关闭] - 关闭成功");
        }
    }

}
